package com.share.config.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

/**
 * @author syq
 * @data 2019/6/17
 */
@Slf4j
@Component
public class KafkaSender {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public SendVo send(String topic, String message) {
        SendVo sendVo = new SendVo();
        try {
            SendResult<String, String> res = kafkaTemplate.send(topic, message).get();
            RecordMetadata recordMetadata = res.getRecordMetadata();
            sendVo.setTopic(recordMetadata.topic());
            sendVo.setPartition(recordMetadata.partition());
            sendVo.setTimestamp(recordMetadata.timestamp());
            sendVo.setOffset(recordMetadata.offset());
            sendVo.setSuccess(true);
        } catch (Exception e) {
            sendVo.setSuccess(false);
            log.error("发送消息至kafka异常-topic={},message={}", topic, message, e);
        }
        return sendVo;
    }
}